package com.cml.infrastructure;

import com.cml.common.config.Configuration;
import com.cml.common.constant.NodeStatus;
import com.cml.common.dto.entity.Message;
import com.cml.common.serializable.MessageSerializable;
import com.cml.domain.service.ClientManagerService;
import com.cml.domain.entity.Session;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * 姓名：陈茂林
 * 时间：2023/10/16 13:58
 * 描述：
 */
public class ClientNetworkManager extends  Thread {
    private ClientManagerService clientManagerService;
    public ClientNetworkManager(ClientManagerService clientManagerService){
        this.clientManagerService=clientManagerService;
    }

    Logger logger = Logger.getLogger(ClientNetworkManager.class);
    @Override
    public void run() {
        ServerSocketChannel server =null;
        try {
            InetSocketAddress socketAddress = new InetSocketAddress(Configuration.getInstance().getServiceManagerPort());
            Selector selector = Selector.open();
            server = ServerSocketChannel.open();
            server.socket().setReuseAddress(true);
            server.socket().bind(socketAddress);
            server.configureBlocking(false);
            server.register(selector, SelectionKey.OP_ACCEPT);
            logger.info("服务管理端口:"+Configuration.getInstance().getServiceManagerPort()+"已启动！");
            while (!server.socket().isClosed()) {
                try {
                    selector.select(1000);

                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();

                        iterator.remove();
                        try {
                            if (key.isAcceptable()) {
                                ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
                                SocketChannel socketChannel = serverChannel.accept();
                                socketChannel.configureBlocking(false);
                                SelectionKey clientKey = socketChannel.register(selector, SelectionKey.OP_READ|SelectionKey.OP_WRITE);
                                Session session = new Session();
                                clientKey.attach(session);
                                ClientQueueManage.createUploadMessageQueue(session.getSessionId());
                                ClientQueueManage.createSendMessageQueue(session.getSessionId());
                                ClientMessageListener clientIoThread = new ClientMessageListener(session,clientManagerService);
                                clientIoThread.start();
                                session.setClientIOThread(clientIoThread);
                                logger.info(socketChannel.getRemoteAddress() + "进行连接");

                            }
                            if (key.isReadable()) {
                                SocketChannel socketChannel = (SocketChannel) key.channel();
                                Session session = (Session) key.attachment();
                                if(session != null) {
                                    Message message = MessageSerializable.decoder(socketChannel);
                                    ClientQueueManage.pushUploadMessageMessage(session.getSessionId(), message);
                                }
                            }
                            if(key.isWritable()){
                                SocketChannel socketChannel = (SocketChannel) key.channel();
                                Session session = (Session) key.attachment();
                                if(session != null){
                                    Message responseMessage = ClientQueueManage.peekSendMessageMessage(session.getSessionId());
                                    if(responseMessage != null){
                                        logger.info("发送到："+socketChannel.getRemoteAddress() +" 信息"+responseMessage);
                                        ByteBuffer byteBuffer = MessageSerializable.encoder(responseMessage);
                                        socketChannel.write(byteBuffer);
                                        //解决粘包问题
                                        if(!responseMessage.getMessageBuffer().hasRemaining()){
                                            ClientQueueManage.pollSendMessageMessage(session.getSessionId());
                                        }

                                    }
                                }

                            }
                        }catch(Exception e){
                            logger.error(e);
                            //这里要做队列的释放和线程关闭
                            Session session = (Session)key.attachment();
                            session.setClose(true);
                            ClientMessageListener clientIOThread = (ClientMessageListener) session.getClientIOThread();
                            clientIOThread.setStop(true);
                            ClientQueueManage.removeUploadMessageQueue(session.getSessionId());
                            ClientQueueManage.removeSendMessageQueue(session.getSessionId());
                            key.cancel();
                        }
                    }

                }catch (Exception e){
                    logger.error(e);
                }
            }
        }catch (Exception e){
            logger.error(e);

        }finally {
            if(server != null){
                try {
                    server.close();
                } catch (IOException e) {
                    logger.error(e);
                    NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
                }
            }
        }
    }
}
